package com.td.module.equipment.framework.core;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

/**
 * @author zhanghe
 * @date 2023/4/16
 */
@Slf4j
public class MqttClientUtil {

    private static MqttClient getMqttClient(String mqttClientId) {
        return MqttClientBuilder.mqttClients.get(mqttClientId);
    }

    /**
     * 订阅某个主题
     *
     * @param topic 主题
     * @param qos   连接方式
     */
    public static void subscribe(String mqttClientId, String topic, int qos) {
        log.info("==============开始订阅主题【{}】==============", topic);
        try {
            getMqttClient(mqttClientId).subscribe(topic, qos);
        } catch (MqttException exception) {
            log.error("订阅出错！");
            throw new RuntimeException(exception);
        }
    }

    /**
     * 订阅多个主题
     *
     * @param topics 主题
     */
    public static void subscribe(String mqttClientId, String[] topics) {
        log.info("==============开始订阅多个主题【{}】==============", (Object) topics);
        try {
            getMqttClient(mqttClientId).subscribe(topics);
        } catch (MqttException exception) {
            log.error("订阅出错！");
            throw new RuntimeException(exception);
        }
    }

    /**
     * 取消订阅某个或多个主题
     *
     * @param topic
     */
    public static void unsubscribe(String mqttClientId, String ...topic) {
        log.info("==============开始取消订阅主题【{}】==============", (Object) topic);
        try {
            getMqttClient(mqttClientId).unsubscribe(topic);
        } catch (MqttException exception) {
            log.error("取消订阅出错！");
            throw new RuntimeException(exception);
        }
    }

    /**
     * 向某个主题发布消息 默认qos：1
     *
     * @param topic:发布的主题
     * @param msg：发布的消息
     */
    public static void publish(String mqttClientId, String topic, String msg) {
        try {
            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setPayload(msg.getBytes());
            MqttTopic mqttTopic = getMqttClient(mqttClientId).getTopic(topic);
            MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
            token.waitForCompletion();
        } catch (MqttException exception) {
            log.error("发布消息出错！");
            throw new RuntimeException(exception);
        }
    }

    /**
     * 向某个主题发布消息
     *
     * @param topic: 发布的主题
     * @param msg:   发布的消息
     * @param qos:   消息质量    Qos：0、1、2
     */
    public static void publish(String mqttClientId, String topic, String msg, int qos) {

        try {

            MqttMessage mqttMessage = new MqttMessage();
            mqttMessage.setQos(qos);
            mqttMessage.setPayload(msg.getBytes());
            MqttTopic mqttTopic = getMqttClient(mqttClientId).getTopic(topic);
            MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
            token.waitForCompletion();

        } catch (MqttException exception) {
            log.error("发布消息出错！");
            throw new RuntimeException(exception);

        }
    }
}
